package com.ks.winterwarmimage.mqtt.service;

import android.app.Service;
import android.content.Context;
import android.os.Bundle;
import android.os.PowerManager;
import android.util.Log;

import org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;

import java.io.File;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * <p>
 * MqttConnection拥有一个MqttAsyncClient{主机、端口、客户端ID}实例，用于对MQTT代理执行MQTT操作。
 * </p>
 * <p>
 * 这里的大多数主要API旨在实现IMqttAsyncClient中最通用形式的方法，并对Android环境进行轻微调整<br>
 * 这些调整通常包括为每种方法添加两个参数 :-
 * <ul>
 * <li>invocationContext - 从应用程序传递的字符串，用于标识操作的上下文（主要用于支持javascript API实现）</li>
 * <li>activityToken - 从Activity传递的字符串，用于关联回回调方法或其他特定于上下文的数据</li>
 * </ul>
 * </p>
 * <p>
 * 操作非常异步，因此通过将相关数据打包到Intent对象中来通知成功和失败，
 * Intent对象通过{@link MqttService#callbackToActivity(String, Status, Bundle)}方法广播回Activity。
 * </p>
 */
class MqttConnection implements MqttCallbackExtended {

    // 意图字符串等..
    private static final String TAG = "MqttConnection";
    // 错误状态消息
    private static final String NOT_CONNECTED = "not connected";

    // 连接定义的字段
    private String serverURI;
    public String getServerURI() {
        return serverURI;
    }

    public void setServerURI(String serverURI) {
        this.serverURI = serverURI;
    }

    public String getClientId() {
        return clientId;
    }

    public void setClientId(String clientId) {
        this.clientId = clientId;
    }

    private String clientId;
    private MqttClientPersistence persistence = null;
    private MqttConnectOptions connectOptions;

    public MqttConnectOptions getConnectOptions() {
        return connectOptions;
    }

    public void setConnectOptions(MqttConnectOptions connectOptions) {
        this.connectOptions = connectOptions;
    }

    // 客户端句柄，用于回调...
    private String clientHandle;

    public String getClientHandle() {
        return clientHandle;
    }

    public void setClientHandle(String clientHandle) {
        this.clientHandle = clientHandle;
    }

    //存储connect ActivityToken以重新连接
    private String reconnectActivityToken = null;

    // 我们的客户端对象-在connect上实例化
    private MqttAsyncClient myClient = null;

    private AlarmPingSender alarmPingSender = null;

    // 我们的（父）服务对象
    private MqttService service = null;

    private volatile boolean disconnected = true;
    private boolean cleanSession = true;

    // 指示此连接是否正在连接。
    // 此变量用于避免多次重新连接。
    private volatile boolean isConnecting = false;

    // 保存已发送的消息及其相应的Topics、activityTokens和invocationContexts，以便我们可以处理来自mqttClient的“deliveryComplete”回调
    private Map<IMqttDeliveryToken, String /* Topic */> savedTopics = new HashMap<>();
    private Map<IMqttDeliveryToken, MqttMessage> savedSentMessages = new HashMap<>();
    private Map<IMqttDeliveryToken, String> savedActivityTokens = new HashMap<>();
    private Map<IMqttDeliveryToken, String> savedInvocationContexts = new HashMap<>();

    private PowerManager.WakeLock wakelock = null;
    private String wakeLockTag = null;

    private DisconnectedBufferOptions bufferOpts = null;

    /**
     * 构造函数-创建MqttConnection以与MQTT服务器通信
     *
     * @param service 我们的“父”服务-我们对其进行回调
     * @param serverURI  我们将连接到的MQTT服务器的URI
     * @param clientId 我们向MQTT服务器标识自己的名称
     * @param persistence 用于存储飞行中消息的持久化类。如果为null，则使用默认的持久性机制
     * @param clientHandle 活动识别我们的“句柄”
     */
    MqttConnection(MqttService service, String serverURI, String clientId,
                   MqttClientPersistence persistence, String clientHandle) {
        this.serverURI = serverURI;
        this.service = service;
        this.clientId = clientId;
        this.persistence = persistence;
        this.clientHandle = clientHandle;

        StringBuilder stringBuilder = new StringBuilder(this.getClass().getCanonicalName());
        stringBuilder.append(" ");
        stringBuilder.append(clientId);
        stringBuilder.append(" ");
        stringBuilder.append("on host ");
        stringBuilder.append(serverURI);
        wakeLockTag = stringBuilder.toString();
    }

    // The major API implementation follows
    /**
     * 连接到实例化时指定的服务器
     *
     * @param options 超时等
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken  要传递回“活动”的任意标识符
     */
    public void connect(MqttConnectOptions options, String invocationContext,
                        String activityToken) {

        connectOptions = options;
        reconnectActivityToken = activityToken;

        if (options != null) {
            cleanSession = options.isCleanSession();
        }

        if (connectOptions.isCleanSession()) { // if it's a clean session,
            // discard old data
            service.messageStore.clearArrivedMessages(clientHandle);
        }

        service.traceDebug(TAG, "Connecting {" + serverURI + "} as {" + clientId + "}");
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,MqttServiceConstants.CONNECT_ACTION);


        try {
            if (persistence == null) {
                // ask Android where we can put files
                File myDir = service.getExternalFilesDir(TAG);

                if (myDir == null) {
                    // No external storage, use internal storage instead.
                    myDir = service.getDir(TAG, Context.MODE_PRIVATE);

                    if(myDir == null){
                        //Shouldn't happen.
                        resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                                "Error! No external and internal storage available");
                        resultBundle.putSerializable(MqttServiceConstants.CALLBACK_EXCEPTION, new MqttPersistenceException());
                        service.callbackToActivity(clientHandle, Status.ERROR,
                                resultBundle);
                        return;
                    }
                }

                // use that to setup MQTT client persistence storage
                persistence = new MqttDefaultFilePersistence(
                        myDir.getAbsolutePath());
            }

            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle) {

                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    doAfterConnectSuccess(resultBundle);
                    service.traceDebug(TAG, "connect success!");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken,
                                      Throwable exception) {
                    resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                            exception.getLocalizedMessage());
                    resultBundle.putSerializable(MqttServiceConstants.CALLBACK_EXCEPTION, exception);
                    service.traceError(TAG,
                            "connect fail, call connect to reconnect.reason:"
                                    + exception.getMessage());

                    doAfterConnectFail(resultBundle);

                }
            };

            if (myClient != null) {
                if (isConnecting ) {
                    service.traceDebug(TAG,
                            "myClient != null and the client is connecting. Connect return directly.");
                    service.traceDebug(TAG,"Connect return:isConnecting:"+isConnecting+".disconnected:"+disconnected);
                }else if(!disconnected){
                    service.traceDebug(TAG,"myClient != null and the client is connected and notify!");
                    doAfterConnectSuccess(resultBundle);
                }
                else {
                    service.traceDebug(TAG, "myClient != null and the client is not connected");
                    service.traceDebug(TAG,"Do Real connect!");
                    setConnectingState(true);
                    myClient.connect(connectOptions, invocationContext, listener);
                }
            }

            // if myClient is null, then create a new connection
            else {
                alarmPingSender = new AlarmPingSender(service);
                myClient = new MqttAsyncClient(serverURI, clientId,persistence, alarmPingSender);
                myClient.setCallback(this);

                service.traceDebug(TAG,"Do Real connect!");
                setConnectingState(true);
                myClient.connect(connectOptions, invocationContext, listener);
            }
        } catch (Exception e) {
            service.traceError(TAG, "Exception occurred attempting to connect: " + e.getMessage());
            setConnectingState(false);
            handleException(resultBundle, e);
        }
    }

    private void doAfterConnectSuccess(final Bundle resultBundle) {
        //since the device's cpu can go to sleep, acquire a wakelock and drop it later.
        acquireWakeLock();
        service.callbackToActivity(clientHandle, Status.OK, resultBundle);
        deliverBacklog();
        setConnectingState(false);
        disconnected = false;
        releaseWakeLock();
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.CONNECT_EXTENDED_ACTION);
        resultBundle.putBoolean(MqttServiceConstants.CALLBACK_RECONNECT, reconnect);
        resultBundle.putString(MqttServiceConstants.CALLBACK_SERVER_URI, serverURI);
        service.callbackToActivity(clientHandle, Status.OK, resultBundle);
    }

    private void doAfterConnectFail(final Bundle resultBundle){
        //
        acquireWakeLock();
        disconnected = true;
        setConnectingState(false);
        service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        releaseWakeLock();
    }

    private void handleException(final Bundle resultBundle, Exception e) {
        resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                e.getLocalizedMessage());

        resultBundle.putSerializable(MqttServiceConstants.CALLBACK_EXCEPTION, e);

        service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
    }

    /**
     * 尝试传递我们收到但应用程序尚未确认的任何未完成消息。如果指定了“cleanSession”，我们将已经从messageStore中清除了所有此类消息。
     */
    private void deliverBacklog() {
        Iterator<MessageStore.StoredMessage> backlog = service.messageStore
                .getAllArrivedMessages(clientHandle);
        while (backlog.hasNext()) {
            MessageStore.StoredMessage msgArrived = backlog.next();
            Bundle resultBundle = messageToBundle(msgArrived.getMessageId(),
                    msgArrived.getTopic(), msgArrived.getMessage());
            resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                    MqttServiceConstants.MESSAGE_ARRIVED_ACTION);
            service.callbackToActivity(clientHandle, Status.OK, resultBundle);
        }
    }

    /**
     * 创建一个包含与消息相关的所有相关数据的包
     *
     * @param messageId  messageStore中的消息标识符，以便在传递后可以进行回调以将其删除
     * @param topic 消息传递的主题
     * @param message 消息本身
     * @return the bundle
     */
    private Bundle messageToBundle(String messageId, String topic,
                                   MqttMessage message) {
        Bundle result = new Bundle();
        result.putString(MqttServiceConstants.CALLBACK_MESSAGE_ID, messageId);
        result.putString(MqttServiceConstants.CALLBACK_DESTINATION_NAME, topic);
        result.putParcelable(MqttServiceConstants.CALLBACK_MESSAGE_PARCEL,
                new ParcelableMqttMessage(message));
        return result;
    }

    /**
     * 关闭服务器连接
     */
    void close() {
        service.traceDebug(TAG, "close()");
        try {
            if (myClient != null) {
                myClient.close();
            }
        } catch (MqttException e) {
            // Pass a new bundle, let handleException stores error messages.
            handleException(new Bundle(), e);
        }
    }

    /**
     * 断开与服务器的连接
     *
     * @param quiesceTimeout 断开超时（以毫秒为单位）
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回活动的任意字符串
     */
    void disconnect(long quiesceTimeout, String invocationContext,
                    String activityToken) {
        service.traceDebug(TAG, "disconnect()");
        disconnected = true;
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.DISCONNECT_ACTION);
        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                myClient.disconnect(quiesceTimeout, invocationContext, listener);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);
            service.traceError(MqttServiceConstants.DISCONNECT_ACTION,
                    NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }

        if (connectOptions != null && connectOptions.isCleanSession()) {
            // assume we'll clear the stored messages at this point
            service.messageStore.clearArrivedMessages(clientHandle);
        }

        releaseWakeLock();
    }

    /**
     * 断开与服务器的连接
     *
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回活动的任意字符串
     */
    void disconnect(String invocationContext, String activityToken) {
        service.traceDebug(TAG, "disconnect()");
        disconnected = true;
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.DISCONNECT_ACTION);
        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                myClient.disconnect(invocationContext, listener);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);
            service.traceError(MqttServiceConstants.DISCONNECT_ACTION,
                    NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }

        if (connectOptions != null && connectOptions.isCleanSession()) {
            // assume we'll clear the stored messages at this point
            service.messageStore.clearArrivedMessages(clientHandle);
        }
        releaseWakeLock();
    }

    /**
     * @return 如果我们连接到MQTT服务器，则为true
     */
    public boolean isConnected() {
        return myClient != null && myClient.isConnected();
    }

    /**
     * 发布关于某个主题的消息
     *
     * @param topic 要发布的主题-表示为字符串，而不是MqttTopic对象
     * @param payload 要发布的消息内容
     * @param qos 所要求的服务质量
     * @param retained MQTT服务器是否应保留此消息
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回活动的任意字符串
     * @return 用于跟踪操作的令牌
     */
    public IMqttDeliveryToken publish(String topic, byte[] payload, int qos,
                                      boolean retained, String invocationContext, String activityToken) {
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.SEND_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);

        IMqttDeliveryToken sendToken = null;

        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                MqttMessage message = new MqttMessage(payload);
                message.setQos(qos);
                message.setRetained(retained);
                sendToken = myClient.publish(topic, payload, qos, retained,
                        invocationContext, listener);
                storeSendDetails(topic, message, sendToken, invocationContext,
                        activityToken);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);
            service.traceError(MqttServiceConstants.SEND_ACTION, NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }

        return sendToken;
    }

    /**
     * 发布关于某个主题的消息
     *
     * @param topic 要发布的主题-表示为字符串，而不是MqttTopic对象
     * @param message 要发布的消息
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回活动的任意字符串
     * @return 用于跟踪操作的令牌
     */
    public IMqttDeliveryToken publish(String topic, MqttMessage message,
                                      String invocationContext, String activityToken) {
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.SEND_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);

        IMqttDeliveryToken sendToken = null;

        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                sendToken = myClient.publish(topic, message, invocationContext,
                        listener);
                storeSendDetails(topic, message, sendToken, invocationContext,
                        activityToken);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else if ((myClient !=null) && (this.bufferOpts != null) && (this.bufferOpts.isBufferEnabled())){
            // Client is not connected, but buffer is enabled, so sending message
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                sendToken = myClient.publish(topic, message, invocationContext,
                        listener);
                storeSendDetails(topic, message, sendToken, invocationContext,
                        activityToken);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        }  else {
            Log.i(TAG, "Client is not connected, so not sending message");
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);
            service.traceError(MqttServiceConstants.SEND_ACTION, NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }
        return sendToken;
    }

    /**
     * 订阅主题
     *
     * @param topic 主题名称（可能是通配符）
     * @param qos 该主题所要求的服务质量
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回“活动”的任意标识符
     */
    public void subscribe(final String topic, final int qos,
                          String invocationContext, String activityToken) {
        service.traceDebug(TAG, "subscribe({" + topic + "}," + qos + ",{"
                + invocationContext + "}, {" + activityToken + "}");
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.SUBSCRIBE_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);

        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                myClient.subscribe(topic, qos, invocationContext, listener);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);
            service.traceError("subscribe", NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }
    }

    /**
     * 订阅一个或多个主题
     *
     * @param topic 主题名称列表（可能带有通配符的）
     * @param qos 每个主题所要求的服务质量
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回“活动”的任意标识符
     */
    public void subscribe(final String[] topic, final int[] qos,
                          String invocationContext, String activityToken) {
        service.traceDebug(TAG, "subscribe({" + Arrays.toString(topic) + "}," + Arrays.toString(qos) + ",{"
                + invocationContext + "}, {" + activityToken + "}");
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.SUBSCRIBE_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);

        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                myClient.subscribe(topic, qos, invocationContext, listener);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);
            service.traceError("subscribe", NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }
    }

    public void subscribe(String[] topicFilters, int[] qos, String invocationContext, String activityToken, IMqttMessageListener[] messageListeners) {
        service.traceDebug(TAG, "subscribe({" + Arrays.toString(topicFilters) + "}," + Arrays.toString(qos) + ",{"
                + invocationContext + "}, {" + activityToken + "}");
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION, MqttServiceConstants.SUBSCRIBE_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN, activityToken);
        resultBundle.putString(MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT, invocationContext);
        if((myClient != null) && (myClient.isConnected())){
            IMqttActionListener listener = new MqttConnectionListener(resultBundle);
            try {

                myClient.subscribe(topicFilters, qos,messageListeners);
            } catch (Exception e){
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE, NOT_CONNECTED);
            service.traceError("subscribe", NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }
    }

    /**
     * 取消订阅某个主题
     *
     * @param topic 主题名称（可能是通配符）
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回“活动”的任意标识符
     */
    void unsubscribe(final String topic, String invocationContext,
                     String activityToken) {
        service.traceDebug(TAG, "unsubscribe({" + topic + "},{"
                + invocationContext + "}, {" + activityToken + "})");
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.UNSUBSCRIBE_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);
        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                myClient.unsubscribe(topic, invocationContext, listener);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);

            service.traceError("subscribe", NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }
    }

    /**
     * 取消订阅一个或多个主题
     *
     * @param topic 主题名称列表（可能带有通配符）
     * @param invocationContext 要传递回应用程序的任意数据
     * @param activityToken 要传递回“活动”的任意标识符
     */
    void unsubscribe(final String[] topic, String invocationContext,
                     String activityToken) {
        service.traceDebug(TAG, "unsubscribe({" + Arrays.toString(topic) + "},{"
                + invocationContext + "}, {" + activityToken + "})");
        final Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.UNSUBSCRIBE_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                activityToken);
        resultBundle.putString(
                MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                invocationContext);
        if ((myClient != null) && (myClient.isConnected())) {
            IMqttActionListener listener = new MqttConnectionListener(
                    resultBundle);
            try {
                myClient.unsubscribe(topic, invocationContext, listener);
            } catch (Exception e) {
                handleException(resultBundle, e);
            }
        } else {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    NOT_CONNECTED);

            service.traceError("subscribe", NOT_CONNECTED);
            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }
    }

    /**
     * 为客户获取所有未完成交付的令牌
     *
     * @return 令牌数组（可能为空）
     */
    public IMqttDeliveryToken[] getPendingDeliveryTokens() {
        return myClient.getPendingDeliveryTokens();
    }

    // Implement MqttCallback
    /**
     * 回调连接丢失
     *
     * @param why 导致连接中断的意外情况
     */
    @Override
    public void connectionLost(Throwable why) {
        service.traceDebug(TAG, "connectionLost(" + why.getMessage() + ")");
        disconnected = true;
        try {
            if(!this.connectOptions.isAutomaticReconnect()) {
                myClient.disconnect(null, new IMqttActionListener() {

                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        // No action
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken,
                                          Throwable exception) {
                        // No action
                    }
                });
            } else {
                // Using the new Automatic reconnect functionality.
                // We can't force a disconnection, but we can speed one up
                alarmPingSender.schedule(100);

            }
        } catch (Exception e) {
            // ignore it - we've done our best
        }

        Bundle resultBundle = new Bundle();
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.ON_CONNECTION_LOST_ACTION);
        if (why != null) {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    why.getMessage());
            if (why instanceof MqttException) {
                resultBundle.putSerializable(
                        MqttServiceConstants.CALLBACK_EXCEPTION, why);
            }
            resultBundle.putString(
                    MqttServiceConstants.CALLBACK_EXCEPTION_STACK,
                    Log.getStackTraceString(why));
        }
        service.callbackToActivity(clientHandle, Status.OK, resultBundle);
        // client has lost connection no need for wake lock
        releaseWakeLock();
    }

    /**
     *回调标识消息已送达（“已送达”的确切含义取决于QOS值）
     *
     * @param messageToken 消息最初发送时提供的mesage令牌
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken messageToken) {

        service.traceDebug(TAG, "deliveryComplete(" + messageToken + ")");

        MqttMessage message = savedSentMessages.remove(messageToken);
        if (message != null) { // If I don't know about the message, it's
            // irrelevant
            String topic = savedTopics.remove(messageToken);
            String activityToken = savedActivityTokens.remove(messageToken);
            String invocationContext = savedInvocationContexts
                    .remove(messageToken);

            Bundle resultBundle = messageToBundle(null, topic, message);
            if (activityToken != null) {
                resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                        MqttServiceConstants.SEND_ACTION);
                resultBundle.putString(
                        MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                        activityToken);
                resultBundle.putString(
                        MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT,
                        invocationContext);

                service.callbackToActivity(clientHandle, Status.OK,
                        resultBundle);
            }
            resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                    MqttServiceConstants.MESSAGE_DELIVERED_ACTION);
            service.callbackToActivity(clientHandle, Status.OK, resultBundle);
        }

        // this notification will have kept the connection alive but send the previously sechudled ping anyway
    }

    /**
     * 收到消息时回调
     *
     * @param topic  收到消息的主题
     * @param message 消息自身
     */
    @Override
    public void messageArrived(String topic, MqttMessage message)
            throws Exception {

        service.traceDebug(TAG,
                "messageArrived(" + topic + ",{" + message.toString() + "})");

        String messageId = service.messageStore.storeArrived(clientHandle,
                topic, message);

        Bundle resultBundle = messageToBundle(messageId, topic, message);
        resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                MqttServiceConstants.MESSAGE_ARRIVED_ACTION);
        resultBundle.putString(MqttServiceConstants.CALLBACK_MESSAGE_ID,
                messageId);
        service.callbackToActivity(clientHandle, Status.OK, resultBundle);

    }



    /**
     * 存储已发送消息的详细信息，以便我们可以处理来自mqttClient的“deliveryComplete”回调
     *
     * @param topic
     * @param msg
     * @param messageToken
     * @param invocationContext
     * @param activityToken
     */
    private void storeSendDetails(final String topic, final MqttMessage msg,
                                  final IMqttDeliveryToken messageToken,
                                  final String invocationContext, final String activityToken) {
        savedTopics.put(messageToken, topic);
        savedSentMessages.put(messageToken, msg);
        savedActivityTokens.put(messageToken, activityToken);
        savedInvocationContexts.put(messageToken, invocationContext);
    }

    /**
     * 为此客户端获取部分唤醒锁
     */
    private void acquireWakeLock() {
        if (wakelock == null) {
            PowerManager pm = (PowerManager) service
                    .getSystemService(Service.POWER_SERVICE);
            wakelock = pm.newWakeLock(PowerManager.PARTIAL_WAKE_LOCK,
                    wakeLockTag);
        }
        wakelock.acquire();

    }

    /**
     * 释放此客户端当前持有的唤醒锁
     */
    private void releaseWakeLock() {
        if(wakelock != null && wakelock.isHeld()){
            wakelock.release();
        }
    }



    /**
     * 客户端上下文的通用IMqttActionListene
     * <p>
     * 简单地处理不返回结果的操作的基本成功/失败案例
     *
     */
    private class MqttConnectionListener implements IMqttActionListener {

        private final Bundle resultBundle;

        private MqttConnectionListener(Bundle resultBundle) {
            this.resultBundle = resultBundle;
        }

        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
            service.callbackToActivity(clientHandle, Status.OK, resultBundle);
        }

        @Override
        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
            resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                    exception.getLocalizedMessage());

            resultBundle.putSerializable(
                    MqttServiceConstants.CALLBACK_EXCEPTION, exception);

            service.callbackToActivity(clientHandle, Status.ERROR, resultBundle);
        }
    }

    /**
     * 收到离线的通知<br>
     * 如果cleanSession为真，我们需要将其视为断开连接
     */
    void offline() {

        if (!disconnected && !cleanSession) {
            Exception e = new Exception("Android offline");
            connectionLost(e);
        }
    }

    /**
     * 重新连接<br>
     * 仅当cleanSession为false并且我们已连接时才适用。
     * 声明为synchronized，以避免多次调用此方法发送connect
     */
    synchronized void reconnect() {

        if (myClient == null) {
            service.traceError(TAG,"Reconnect myClient = null. Will not do reconnect");
            return;
        }

        if (isConnecting) {
            service.traceDebug(TAG, "The client is connecting. Reconnect return directly.");
            return ;
        }

        if(!service.isOnline()){
            service.traceDebug(TAG,
                    "The network is not reachable. Will not do reconnect");
            return;
        }

        if(connectOptions.isAutomaticReconnect()){
            //The Automatic reconnect functionality is enabled here
            Log.i(TAG, "Requesting Automatic reconnect using New Java AC");
            final Bundle resultBundle = new Bundle();
            resultBundle.putString(
                    MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                    reconnectActivityToken);
            resultBundle.putString(
                    MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT, null);
            resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                    MqttServiceConstants.CONNECT_ACTION);
            try {
                myClient.reconnect();
            } catch (MqttException ex){
                Log.e(TAG, "Exception occurred attempting to reconnect: " + ex.getMessage());
                setConnectingState(false);
                handleException(resultBundle, ex);
            }
        } else if (disconnected && !cleanSession) {
            // use the activityToke the same with action connect
            service.traceDebug(TAG,"Do Real Reconnect!");
            final Bundle resultBundle = new Bundle();
            resultBundle.putString(
                    MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN,
                    reconnectActivityToken);
            resultBundle.putString(
                    MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT, null);
            resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION,
                    MqttServiceConstants.CONNECT_ACTION);

            try {

                IMqttActionListener listener = new MqttConnectionListener(resultBundle) {
                    @Override
                    public void onSuccess(IMqttToken asyncActionToken) {
                        // since the device's cpu can go to sleep, acquire a
                        // wakelock and drop it later.
                        service.traceDebug(TAG,"Reconnect Success!");
                        service.traceDebug(TAG,"DeliverBacklog when reconnect.");
                        doAfterConnectSuccess(resultBundle);
                    }

                    @Override
                    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                        resultBundle.putString(
                                MqttServiceConstants.CALLBACK_ERROR_MESSAGE,
                                exception.getLocalizedMessage());
                        resultBundle.putSerializable(
                                MqttServiceConstants.CALLBACK_EXCEPTION,
                                exception);
                        service.callbackToActivity(clientHandle, Status.ERROR,
                                resultBundle);

                        doAfterConnectFail(resultBundle);

                    }
                };

                myClient.connect(connectOptions, null, listener);
                setConnectingState(true);
            } catch (MqttException e) {
                service.traceError(TAG, "Cannot reconnect to remote server." + e.getMessage());
                setConnectingState(false);
                handleException(resultBundle, e);
            } catch (Exception e){
				/*  TODO: Added Due to: https://github.com/eclipse/paho.mqtt.android/issues/101
				    For some reason in a small number of cases, myClient is null here and so
				    a NullPointer Exception is thrown. This is a workaround to pass the exception
				    up to the application. myClient should not be null so more investigation is
				    required.
				*/
                service.traceError(TAG, "Cannot reconnect to remote server." + e.getMessage());
                setConnectingState(false);
                MqttException newEx = new MqttException(MqttException.REASON_CODE_UNEXPECTED_ERROR, e.getCause());
                handleException(resultBundle, newEx);
            }
        }
    }

    /**
     *
     * @param isConnecting
     */
    private synchronized void setConnectingState(boolean isConnecting){
        this.isConnecting = isConnecting;
    }

    /**
     * 为此客户端设置DisconnectedBufferOptions
     * @param bufferOpts
     */
    public void setBufferOpts(DisconnectedBufferOptions bufferOpts) {
        this.bufferOpts = bufferOpts;
        myClient.setBufferOpts(bufferOpts);
    }

    public int getBufferedMessageCount(){
        return myClient.getBufferedMessageCount();
    }

    public MqttMessage getBufferedMessage(int bufferIndex){
        return myClient.getBufferedMessage(bufferIndex);
    }

    public void deleteBufferedMessage(int bufferIndex){
        myClient.deleteBufferedMessage(bufferIndex);
    }
}
